package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @author Felix
 * @date 2022/11/12
 * 专门用于查询维度的工具类
 */
public class DimUtil {

    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("id",id));
    }
    /**
     * 使用旁路缓存进行维度查询
     * 思路：先从缓存中获取维度数据，如果缓存中找到了对应的维度数据，直接将缓存的维度进行返回（缓存命中）;
     * 如果在缓存中没有找到对应的维度数据，那么发送请求到Phoenix表中去查询维度，并将查询到的维度放到缓存中，方便下次查询使用
     * 缓存产品的选择
     * Redis：性能好、维护性好  √
     * 状态:性能好、维护性差
     * Redis的相关配置
     * Key：     dim:维度表名:主键1_主键2
     * 类型:     String
     * ex：     1day
     * 注意：如果业务数据库维度表发生了变化，删除缓存中的维度数据
     *
     * @param tableName
     * @param columnNameAndValues
     * @return
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接从Redis中查询维度的key
        StringBuilder redisKey = new StringBuilder("dim:" + tableName.toLowerCase() + ":");
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            redisKey.append(columnValue);
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String dimInfoStr = null;
        JSONObject dimJsonObj = null;

        try {
            //先从Redis中查询维度数据
            jedis = RedisUtil.getJedis();
            dimInfoStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            System.out.println("~~从Redis中查询维度数据发生了异常~~");
        }

        //判断Redis中是否存在维度数据
        if (StringUtils.isNotEmpty(dimInfoStr)) {
            //缓存命中，将缓存的维度字符串转换为json对象 进行返回
            dimJsonObj = JSON.parseObject(dimInfoStr);
        } else {
            //如果在Redis中没有找到对应的维度数据，那么发送请求到Phoenix表中进行查询
            System.out.println("从Phoenix表中查询维度数据的SQL:" + selectSql);
            //底层还是调用的PhoenixUtil中queryList方法
            List<JSONObject> dimJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if (dimJsonObjList != null && dimJsonObjList.size() > 0) {
                //因为是根据维度的主键 到维度表中查询维度数据  ，如果有数据，也只会有一条，所以将集合中的这条数据取出来，就是我们需要的维度
                dimJsonObj = dimJsonObjList.get(0);
                //将从Phoenix表中查询出来的数据 放到Redis中缓存起来
                if (jedis != null) {
                    jedis.setex(redisKey.toString(), 3600 * 24, dimJsonObj.toJSONString());
                }
            } else {
                System.out.println("~~在Phoenix表中没有找到对应的维度数据~~");
            }
        }

        //释放资源
        if(jedis != null){
            System.out.println("~~关闭Jedis客户端~~");
            jedis.close();
        }

        return dimJsonObj;
    }

    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");

        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
            }
        }
        System.out.println("从Phoenix表中查询维度数据的SQL:" + selectSql);
        //底层还是调用的PhoenixUtil中queryList方法
        List<JSONObject> dimJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimJsonObj = null;
        if (dimJsonObjList != null && dimJsonObjList.size() > 0) {
            //因为是根据维度的主键 到维度表中查询维度数据  ，如果有数据，也只会有一条，所以将集合中的这条数据取出来，就是我们需要的维度
            dimJsonObj = dimJsonObjList.get(0);
        } else {
            System.out.println("~~在Phoenix表中没有找到对应的维度数据~~");
        }

        return dimJsonObj;
    }


    //从Redis中删除维度
    public static void delCached(String tableName,String id){
        String redisKey= "dim:" + tableName.toLowerCase() + ":" + id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("~~从Redis中删除维度数据发生了异常~~");
        } finally {
            if(jedis != null){
                System.out.println("~~从Redis中删除维度后关闭客户端~~");
                jedis.close();
            }
        }
    }

    public static void main(String[] args) {
        System.out.println(DimUtil.getDimInfo("dim_base_trademark", Tuple2.of("id", "1")));
    }
}
